"""
Case Type   : 连接
Case Name   : 并行修改数据库，用户，会话参数，设置idle_in_transaction_session_timeout为20s，40s, 60s，处于idle_in_transaction_block状态观察会话断开时间
Create At   : 2024/10/14
Owner       : @li-xin12345
Description :
    1、修改数据库级别参数，设置idle_in_transaction_session_timeout为20s；
    alter database postgres set idle_in_transaction_session_timeout to 20;
    2、修改用户级别参数，设置idle_in_transaction_session_timeout为40s；
    alter user cy903 set idle_in_transaction_session_timeout to 40;
    3、修改会话级别参数，设置idle_in_transaction_session_timeout为60s；
    set idle_in_transaction_session_timeout to 60;
    4、开启事务，分别等待20s，40s，60s查看数据库的状态
    begin;
    select now();
    show now();
    等待21s
    select now();
    等待41s
    select now();
    等待61s
    select now();
    end;
    5、清理环境
    alter database postgres set idle_in_transaction_session_timeout to 0;
    alter user cy903 set idle_in_transaction_session_timeout to 0;
Expect      :
    1、成功
    2、成功
    3、成功
    4、分别等待20s数据库处于连接状态，等待40s数据库处于连接状态，等待60s数据库处于断开状态；
    5、成功
History     :
"""
import os
import time
import unittest

from yat.test import Node
from yat.test import macro

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from testcase.utils.ComThread import ComThread

pri_sh = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 != pri_sh.get_node_num(), '非1+2环境不执行')
class SessionTest(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        text = f'-----{os.path.basename(__file__)} start-----'
        self.log.info(text)
        self.constant = Constant()
        self.common = Common()
        self.pri_root_node = Node('PrimaryRoot')
        self.pri_user_node = Node('PrimaryDbUser')
        self.db_name = 'db_connect_case0353'
        self.user_name = 'u_connect_case0353'
        self.script_name = 'Guc_connectionauthentication_Case0353'
        self.dest_path = os.path.join(
            macro.DB_MY_PATH, 'guc_connect_Case0353').replace('\\', '/')

        self.log.info('--前置处理：创建数据库--')
        cmd = f"drop database if exists {self.db_name};" \
              f"create database {self.db_name};"
        res = pri_sh.execut_db_sql(cmd)
        self.log.info(res)
        self.assertIn(self.constant.DROP_DATABASE_SUCCESS, res)
        self.assertIn(self.constant.CREATE_DATABASE_SUCCESS, res)

        self.log.info('--前置处理：创建用户--')
        cmd = f"drop user if exists {self.user_name} cascade;" \
              f"create user {self.user_name} " \
              f"password '{self.pri_user_node.db_password}';" \
              f"grant all privileges to {self.user_name};"
        res = pri_sh.execut_db_sql(cmd)
        self.log.info(res)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG, res,
                      f'执行失败: {text}')
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG, res,
                      f'执行失败: {text}')
        self.assertIn(self.constant.ALTER_ROLE_SUCCESS_MSG, res,
                      f'执行失败: {text}')

    def test_connect(self):
        text = '----step1: 修改数据库级别参数，' \
               '设置idle_in_transaction_session_timeout为20s；expect: 成功----'
        self.log.info(text)
        cmd = f"alter database {self.db_name} " \
              f"set idle_in_transaction_session_timeout to 20;"
        res = pri_sh.execut_db_sql(cmd)
        self.log.info(res)
        self.assertIn(self.constant.ALTER_DATABASE_SUCCESS_MSG, res,
                      f'执行失败: {text}')

        text = '----step2: 修改用户级别参数，' \
               '设置idle_in_transaction_session_timeout为40s；expect: 成功----'
        self.log.info(text)
        cmd = f"alter user {self.user_name} " \
              f"set idle_in_transaction_session_timeout to 40;"
        res = pri_sh.execut_db_sql(cmd)
        self.log.info(res)
        self.assertIn(self.constant.ALTER_ROLE_SUCCESS_MSG, res, f'执行失败: {text}')
        text = '----step3: 修改会话级别参数，' \
               '设置idle_in_transaction_session_timeout为60s；expect: 成功----'
        self.log.info(text)
        self.log.info('---步骤3、4在java中实现---')
        text = '----step4: 开启事务，其中执行错误语句，' \
               '分别等待20s，40s，60s查看数据库的状态；' \
               'expect: 等待20s数据库处于连接状态，等待40s数据库处于连接状态，' \
               '等待60s数据库处于断开状态；----'
        self.log.info(text)
        self.log.info('---调用java脚本执行步骤3、4---')
        self.log.info('--远程复制java脚本至指定主机--')
        self.common.scp_file(self.pri_root_node,
                             f"{self.script_name}.java", self.dest_path)
        cmd = f'ls {self.dest_path}'
        res = self.pri_root_node.sh(cmd).result()
        self.assertIn(self.script_name, res, '远程复制java脚本至指定主机失败')

        self.log.info('--编译java工具--')
        self.pri_root_node.scp_put(macro.JDBC_PATH,
                                   f"{self.dest_path}/postgresql.jar")
        cmd = f"javac -encoding utf-8 -cp " \
              f"{os.path.join(self.dest_path, 'postgresql.jar')} " \
              f"{os.path.join(self.dest_path, f'{self.script_name}.java')}"
        self.log.info(cmd)
        result = self.pri_root_node.sh(cmd).result()
        self.log.info(result)
        self.assertEqual('', result, '编译java工具失败')

        self.log.info('--运行java脚本--')
        url_host_port = f"{self.pri_user_node.db_host}:" \
                        f"{self.pri_user_node.db_port}"
        cmd = f" java -cp {os.path.join(self.dest_path, 'postgresql.jar')}" \
              f":{self.dest_path} " \
              f"{self.script_name} {url_host_port} " \
              f"{self.db_name} {self.user_name} " \
              f"{self.pri_user_node.db_password} 2>&1"
        self.log.info(cmd)
        sql_thread = ComThread(self.pri_root_node.sh, args=(cmd,))
        sql_thread.setDaemon(True)
        sql_thread.start()
        time.sleep(2)
        query_sql = f"select * from pg_stat_activity " \
                    f"where usename = '{self.user_name}';"
        check_session_exist = pri_sh.execut_db_sql(query_sql,
                                                        dbname=self.db_name)
        self.log.info(check_session_exist)
        self.assertIn('1 row', check_session_exist, '执行失败' + text)
        self.assertIn('idle in transaction (aborted)', check_session_exist, '执行失败' + text)

        self.log.info('--等待20s，查看会话状态--')
        time.sleep(21)
        check_session_exist = pri_sh.execut_db_sql(query_sql,
                                                        dbname=self.db_name)
        self.log.info(check_session_exist)
        self.assertIn('1 row', check_session_exist, '执行失败' + text)

        self.log.info('--再等待20s，查看会话状态--')
        time.sleep(21)
        check_session_exist = pri_sh.execut_db_sql(query_sql,
                                                        dbname=self.db_name)
        self.log.info(check_session_exist)
        self.assertIn('1 row', check_session_exist, '执行失败' + text)

        self.log.info('--再等待20s，查看会话状态--')
        time.sleep(21)
        check_session_exist = pri_sh.execut_db_sql(query_sql,
                                                        dbname=self.db_name)
        self.log.info(check_session_exist)
        self.assertIn('0 rows', check_session_exist, '执行失败' + text)

        sql_thread.join(100)
        res = sql_thread.get_result()
        self.log.info(res)

    def tearDown(self):
        text = '----step5: 清理环境;expect:成功----'
        self.log.info(text)
        text1 = f'----后置操作：恢复idle_in_transaction_session_timeout默认值----'
        self.log.info(text1)
        cmd = f"alter database {self.db_name} " \
              f"set idle_in_transaction_session_timeout to 20;" \
              f"alter user {self.user_name} " \
              f"set idle_in_transaction_session_timeout to 40;"
        res1 = pri_sh.execut_db_sql(cmd)
        self.log.info(res1)

        text2 = f'----后置操作：删除数据库、用户----'
        self.log.info(text2)
        cmd = f'drop database if exists {self.db_name};' \
              f'drop user if exists {self.user_name} cascade;'
        res2 = pri_sh.execut_db_sql(cmd)
        self.log.info(res2)

        self.assertIn(self.constant.ALTER_DATABASE_SUCCESS_MSG, res1,
                      f'执行失败: {text1}')
        self.assertIn(self.constant.ALTER_ROLE_SUCCESS_MSG, res1,
                      f'执行失败: {text1}')
        self.assertIn(self.constant.DROP_DATABASE_SUCCESS, res2,
                      f'执行失败: {text2}')
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG, res2,
                      f'执行失败: {text2}')

        text = f'-----{os.path.basename(__file__)} end-----'
        self.log.info(text)
